package com.atguigu.realtime.app.dwd.db;

import com.atguigu.realtime.app.BaseSQLApp;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2022/7/22 10:22
 */
public class Dwd_10_DwdTradeRefundPaySuc extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_10_DwdTradeRefundPaySuc().init(
            2010,
            2,
            "Dwd_10_DwdTradeRefundPaySuc"
        );
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        // 1. 读取ods_db
        readOdsDb(tEnv, "Dwd_10_DwdTradeRefundPaySuc");
        
        // 2. 读取字典表
        readBaseDic(tEnv);
        
        // 3. 过滤退款成功表  refund_status=0705  模拟数据中没有这个状态, 所以注释掉条件
        Table refundPayment = tEnv
            .sqlQuery("select " +
                          "data['id'] id, " +
                          "data['order_id'] order_id, " +
                          "data['sku_id'] sku_id, " +
                          "data['payment_type'] payment_type, " +
                          "data['callback_time'] callback_time, " +
                          "data['total_amount'] total_amount, " +
                          "data['refund_status'] refund_status, " +
                          "pt, " +
                          "ts " +
                          "from ods_db " +
                          "where `database`='gmall2022' " +
                          "and `table`='refund_payment' "
                          /*+
                          "and `type`='update' " +
                          "and `old`['refund_status'] is not null " +
                          "and `data['refund_status']`='0705'"*/
            );
        tEnv.createTemporaryView("refund_payment", refundPayment);
        // 4. 过滤订单表: 退款成功 order_status=1006
        Table orderInfo = tEnv
            .sqlQuery("select " +
                          "data['id'] id, " +
                          "data['province_id'] province_id, " +
                          "`old` " +
                          "from ods_db " +
                          "where `database`='gmall2022' " +
                          "and `table`='order_info' " +
                          "and `type`='update' " +
                          "and `old`['order_status'] is not null " +
                          "and `data`['order_status']='1006'");
        tEnv.createTemporaryView("order_info", orderInfo);
        // 5. 过滤退单表:  refund_status=0705  模拟数据中没有这个状态, 所以注释掉条件
        Table orderRefundInfo = tEnv
            .sqlQuery("select " +
                          "data['id'] id, " +
                          "data['user_id'] user_id, " +
                          "data['order_id'] order_id, " +
                          "data['sku_id'] sku_id, " +
                          "data['refund_type'] refund_type, " +
                          "data['refund_num'] refund_num, " +
                          "data['refund_amount'] refund_amount, " +
                          "data['refund_reason_type'] refund_reason_type, " +
                          "data['refund_reason_txt'] refund_reason_txt, " +
                          "data['create_time'] create_time, " +
                          "pt, " +
                          "ts " +
                          "from ods_db " +
                          "where `database`='gmall2022' " +
                          "and `table`='order_refund_info' "
                         /* +
                          "and `type`='update'"*/
            );
        tEnv.createTemporaryView("order_refund_info", orderRefundInfo);
        // 6. join
        Table result = tEnv.sqlQuery("select " +
                                         "rp.id, " +
                                         "ori.user_id, " +
                                         "rp.order_id, " +
                                         "rp.sku_id, " +
                                         "oi.province_id, " +
                                         "rp.payment_type, " +
                                         "dic.dic_name payment_type_name, " +
                                         "date_format(rp.callback_time,'yyyy-MM-dd') date_id, " +
                                         "rp.callback_time, " +
                                         "ori.refund_num, " +
                                         "rp.total_amount, " +
                                         "rp.ts, " +
                                         "current_row_timestamp() row_op_ts " +
                                         "from refund_payment rp " +
                                         "join base_dic for system_time as of rp.pt as dic " +
                                         "on rp.payment_type=dic.dic_code " +
                                         "join order_info oi " +
                                         "on rp.order_id=oi.id " +
                                         "join order_refund_info ori " +
                                         "on rp.order_id=ori.order_id and rp.sku_id=ori.sku_id");// 退款表和退单表进行join时, 要用到两个条件: 订单id和sku_id
        
        
        // 7. 写出到Kafka
        tEnv.executeSql("create table dwd_trade_refund_pay_suc( " +
                                 "id string, " +
                                 "user_id string, " +
                                 "order_id string, " +
                                 "sku_id string, " +
                                 "province_id string, " +
                                 "payment_type_code string, " +
                                 "payment_type_name string, " +
                                 "date_id string, " +
                                 "callback_time string, " +
                                 "refund_num string, " +
                                 "refund_amount string, " +
                                 "ts string, " +
                                 "row_op_ts timestamp_ltz(3) " +
                                 ")" + SQLUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_TRADE_REFUND_PAY_SUC));
    
        result.executeInsert("dwd_trade_refund_pay_suc");
    
    }
}
